package edu.gatech.doi.spamframework.mapping;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import weka.core.Attribute;
import weka.core.FastVector;
import weka.core.Instance;
import weka.core.Instances;
import weka.core.converters.ArffSaver;
import edu.gatech.doi.spamframework.core.MappingXMLFiles;
import edu.gatech.doi.spamframework.core.ModelXMLFiles;
import edu.gatech.doi.spamframework.model.ModelAttribute;
import edu.gatech.doi.spamframework.model.XMLOperator;

public class TwitterSpamHamFilter implements Mapper {
	private String modelsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"models"+System.getProperties().getProperty("file.separator");
	
	private String mappingsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"mappings"+System.getProperties().getProperty("file.separator");
	private String propertiesdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"properties"+System.getProperties().getProperty("file.separator");
	private String instancesdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"instances"+System.getProperties().getProperty("file.separator");
	private String datasetsdirectory=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"datasets"+System.getProperties().getProperty("file.separator")+"Twitter"+System.getProperties().getProperty("file.separator");
	
	private String personmodelfilename="personmodel.xml";
	
	private String messagemodelfilename="messagemodel.xml";
	
	private String personmappingfilename="twitterpersonmapping.xml";
	private String messagemappingfilename="twittermessagemapping.xml";
	
	private HashMap personmodelhm;
	
	private HashMap personmappinghm;
	
	private HashMap messagemodelhm;
	
	private HashMap messagemappinghm;
	
	private String twitterpersonarfffilename;
	
	private String twittermessagearfffilename;
	
	private String URL;
	
	private String username;
	
	private String password;

	//0: append; 1: replace
	private int isappended=0;
	
	private HashMap personspamhm=null;
	
	private HashMap messagespamhm=null;
	
	public void init()
	{	
		  //Get the jvm heap size.  

		
		long heapSize = Runtime.getRuntime().totalMemory();  
		
		//Print the jvm heap size.  

	    System.out.println("Heap Size = " + heapSize);  

		
		// load property file for myspace
		
		String filepath0=this.propertiesdirectory+"twitter.properties";
//		System.out.println(filepath0);
		PropertiesFileOperator pfo=new PropertiesFileOperator(filepath0);
		pfo.load();
		HashMap mhm=ModelXMLFiles.getMap();
		HashMap mappinghm=MappingXMLFiles.getMap("TWITTER");
		this.messagemodelfilename=(String) mhm.get("MESSAGE");
		this.personmappingfilename=(String) mappinghm.get("TWITTER.PERSON");
		this.personmodelfilename=(String) mhm.get("PERSON");
		this.twittermessagearfffilename=pfo.getProperty("MESSAGEARFF");
		this.twitterpersonarfffilename=pfo.getProperty("PERSONARFF");
		this.isappended=Integer.parseInt(pfo.getProperty("ISAPPENDED"));
		this.URL=pfo.getProperty("URL");
		this.username=pfo.getProperty("USERNAME");
		this.password=pfo.getProperty("PASSWORD");
		
//		System.out.println(this.twitterpersonarfffilename);
//		this.twitterpersongarfffilename="twitterperson.arff";
//		
//		this.twittermessagearfffilename="twittermessage.arff";
		
//		System.out.println(this.messagemodelfilename+" "+this.personmappingfilename+" "+this.personmodelfilename);
		
		// load models and mappings
		XMLOperator xmlo=new XMLOperator();
		//person models
		String filepath1=this.modelsdirectory+this.personmodelfilename;
		this.personmodelhm=new HashMap();
//		System.out.println("filepath:"+filepath);
		this.personmodelhm=xmlo.getModelAttrsFromXML(filepath1);
//		if(this.personmodelhm==null)
//		{
//			System.out.println("personmappinghm is null");
//		}
		String filepath2=this.mappingsdirectory+this.personmappingfilename;
		this.personmappinghm=new HashMap();
		this.personmappinghm=xmlo.getMappingAttrsFromXML(filepath2);
		
		
		//message models
		String filepath3=this.modelsdirectory+this.messagemodelfilename;
		this.messagemodelhm=new HashMap();
//		System.out.println("filepath:"+filepath);
		this.messagemodelhm=xmlo.getModelAttrsFromXML(filepath3);
//		if(this.personmodelhm==null)
//		{
//			System.out.println("personmappinghm is null");
//		}
		String filepath4=this.mappingsdirectory+this.messagemappingfilename;
		this.messagemappinghm=new HashMap();
		this.messagemappinghm=xmlo.getMappingAttrsFromXML(filepath4);
		
		String personspamfilename=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"twittersenderidL1.txt";
		String messagespamfilename=System.getProperties().getProperty("user.dir")+System.getProperties().getProperty("file.separator")+"suspended_status_ids";
		File personspamfile=new File(personspamfilename);
		File messagespamfile=new File(messagespamfilename);
		this.personspamhm=new HashMap();
		this.messagespamhm=new HashMap();
		try {
			FileReader frs=new FileReader(personspamfile);
			BufferedReader br=new BufferedReader(frs);
			String line=null;
			while((line=br.readLine())!=null)
			{
				this.personspamhm.put(line.trim(), "1");
			}
			br.close();
			frs.close();
			
			FileReader frs2=new FileReader(messagespamfile);
			BufferedReader br2=new BufferedReader(frs2);
			
			while((line=br2.readLine())!=null)
			{
				this.messagespamhm.put(line.trim(), "1");
			}
			br2.close();
			frs2.close();
			
			
			
		} catch (FileNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		

		
	}
	
	
	@Override
	public Instances parsePersons(String personFile) {
		// TODO Auto-generated method stub
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.personmappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.personmappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection(this.URL+
		"?user="+this.username+"&password="+this.password);
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.personmodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.personmodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterPersons", atts, 0);
		    
		    Instances tempdata=data;
		    HashMap hm=this.personMapping(attributelist);
		    
		    
			 //create file
		    String spamorhamfilename="twitterpersonspam.arff";
			    String arfffilename=this.instancesdirectory+spamorhamfilename;   
			    File arfffile=new File(arfffilename);
			    if(!arfffile.exists())
			    {
			    	try {
			    		arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, spamorhamfilename);
		 
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    else
			    {
			    	try {
				    	
							arfffile.delete();
							arfffile.createNewFile();
						 	this.save2arff(tempdata, this.instancesdirectory,spamorhamfilename);
			    		
			    	
		    		} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
			    }
			    
			 
		    
		   
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
		 
		    
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 
			 int missingnum=0;
			 int spamnum=0;
			 int hamnum=0;
			 HashMap inhm=new HashMap();
			 
			 Iterator itspam=this.personspamhm.keySet().iterator();
			 
		 while(itspam.hasNext())
		 {
			
			 String name=(String) itspam.next();
			 System.out.println("SELECT id,name,location,description,url FROM user_info WHERE id="+name.trim()+";");
			 stmt.execute("SELECT id,name,location,description,url FROM user_info where id="+name.trim()+";");
			
			
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		 

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    while(rs.next())
			    {
				    num++;
//			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
			    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
	
						
						
						String rows = null;
						
						rows=rs.getString(attstr.toLowerCase());
//						System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
					
				    String labelfornewinst="missing";
				    
				    for(int i=0;i<data.numAttributes();i++)
					{
						
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								break;
							
							if(data.attribute(i).name().equalsIgnoreCase("id"))
							{
								String key=(String)inhm.get(data.attribute(i).name());
								if(this.personspamhm.containsKey(key))
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
									labelfornewinst="spam";
								}
								else
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
									labelfornewinst="ham";
								}
							}
							
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr="Twitter_"+id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
//					System.out.println(newInst.toString());
				    // add
				    data.add(newInst);
				   
				
			
				    
				    
			
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
//			    if(num==0)
//			    	break;
			 // 4. output data
	//		    System.out.println(data);
			    
			   
			}
			
			
		  
		   
		    // first time saving 
		 
			 //append new instance to the file
	    	
	    	writer =new FileWriter(arfffilename, true);
	    	
		    bw=new BufferedWriter(writer);
		    try {
		    	
		    	  // You are now at the end of the file,
		    	  // and can start writing new data out, e.g.
//		    	 bw.newLine(); 
		    	 for(int i=0;i<data.numInstances();i++)
		    	 {
//		    		 System.out.println(data.instance(i).value(data.numAttributes()-1));
		    	
			    		 bw.newLine();
			    		 bw.write(data.instance(i).toString());
			    		
			    		 bw.flush();
//		    		 }
		    	 }
		    	 bw.close();
		    	 writer.close();
		    	 bw=null;
		    	 writer=null;
		    	} catch (IOException ex ) {
		    	  ex.printStackTrace();
		    	}finally {                       // always close the file
		    		 if (bw != null) try {
		    		    bw.close();
		    		 } catch (IOException ioe2) {
		    		    // just ignore it
		    		 }
		    	 }
		    	// delete instances in data
		    data.delete();
	    	
		  
		   
		   
		    
		    
//		    Runtime.getRuntime().gc();
//			offsetnum=offsetnum+limit;
		 }
		   System.out.println("missingnum="+missingnum);
		   System.out.println("hamnum="+hamnum);
		   System.out.println("spamnum="+spamnum);
		
		 
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}

	@Override
	public Instances parseMessages(String msgFile) {
		// TODO Auto-generated method stub
		
		Instances       data = null;
		FastVector      atts;
		/**
		 * attributes list
		 */
		
		ArrayList attributelist=new ArrayList();
		/**
		 * id name location description  url
		 */
//		attributelist.add("ID");
//		attributelist.add("NAME");
//		attributelist.add("LOCATION");
//		attributelist.add("DESCRIPTION");
//		attributelist.add("URL");
//		
		Iterator attrit=this.messagemappinghm.keySet().iterator();
		HashMap mappingattrhm=new HashMap();
		while(attrit.hasNext())
		{
			String tempattrstr=(String) attrit.next();
			attributelist.add(tempattrstr);
			MappingAttribute mappinga=(MappingAttribute) this.messagemappinghm.get(tempattrstr);
			mappingattrhm.put(mappinga.getModelAttributeName(), mappinga);
		}
		
		/**
		 * connect to db :Mysql
		 */
		
		Connection conn = null;

		try {
		System.out.println("before connecting...");
		conn =DriverManager.getConnection("jdbc:postgresql://matcha.cc.gatech.edu:5432/twitware?" +
		"user=wang6&password=?WDcj984615?");
		System.out.println("connect successfully");
		// Do something with the Connection
		Statement stmt = null;
		ResultSet rs = null;
		try {
		stmt = (Statement) conn.createStatement();
		//rs = stmt.executeQuery("SELECT foo FROM bar");
		// or alternatively, if you don't know ahead of time that
		// the query will be a SELECT...
		
		 stmt.execute("set search_path to twitware,public;");
		 System.out.println("set search_path to twitware,public;");
		 stmt.setFetchSize(100); 
		 
			// 1. set up attributes
			
		    atts = new FastVector();
		    /**
			 * id name location description  url
			 * 
			 */
		   Set personmodelkeyset= this.messagemodelhm.keySet();
		   List list=new ArrayList(personmodelkeyset);
		   Collections.sort(list);
		    Iterator it=list.iterator();
		    while(it.hasNext())
		    {
		    	String tempstr=(String)it.next();
//		    	System.out.println("tempstr1="+tempstr);
		    	ModelAttribute attribute=(ModelAttribute)this.messagemodelhm.get(tempstr.trim());
		    	String attributename=attribute.getAttributeName();
//		    	System.out.println("attributename:"+attributename);
		    	int typeid=attribute.getAttributeTypeId();
		    	
		    	switch(typeid)
		    	{
		    	case 1:
		    		//-numeric
		    		
		    		atts.addElement(new Attribute(attributename));
		    		break;
		    	case 2:
		    		//-nominal
		    		FastVector attVals;
		    	    attVals = new FastVector();
		    	    ArrayList al=attribute.getNomialValues();
		    	    Iterator ital=al.iterator();
		    	    while(ital.hasNext())
		    	    {
		    	    	String alstr=(String) ital.next();
		    	    	attVals.addElement(alstr);
		    	    }
		    	    atts.addElement(new Attribute(attributename, attVals));
		    	    break;
		    		
		    	case 3:
		    		//-string
		    		atts.addElement(new Attribute(attributename, (FastVector) null));
		    		break;
		    	case 4:
		    		//-date
		    	    atts.addElement(new Attribute(attributename, (FastVector) null));
		    	    break;
		    	}
		    }
		 // - string
		    // instance_id and label
		    Attribute instanceid=new Attribute("INSTANCE_ID", (FastVector) null);
		    Attribute label=new Attribute("LABEL", (FastVector) null);
		    atts.addElement(instanceid);
		    atts.addElement(label);
		    
		    
		 // 2. create Instances object
		    data = new Instances("TwitterMessages", atts, 0);
		    Instances tempdata=data;
		    HashMap hm=this.messageMapping(attributelist);
		    
			 //create file
		    String spamorhamfilename="twittermessagespam.arff";
		    String arfffilename=this.instancesdirectory+spamorhamfilename;   
		    File arfffile=new File(arfffilename);
		    if(!arfffile.exists())
		    {
		    	try {
		    		arfffile.createNewFile();
				 	this.save2arff(tempdata, this.instancesdirectory, spamorhamfilename);
	 
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		    }
		    else
		    {
		    	try {
			    	
						arfffile.delete();
						arfffile.createNewFile();
					 	this.save2arff(tempdata, this.instancesdirectory, spamorhamfilename);
		    		
		    	
	    		} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		    }
		    
		    
			
		    
		    
		    
		    FileWriter writer = null;
		    BufferedWriter bw=null;
	    	 
		    writer =new FileWriter(arfffilename, true);
	    	
		    bw=new BufferedWriter(writer);
		    
		 
		    
	    	 int offsetnum = 0;
			 int totalnum=0;
			 int limit=5000;
			 int missingnum=0;
			 int hamnum=0;
			 int spamnum=0;
			 HashMap inhm=new HashMap();
			 Iterator it2=this.messagespamhm.keySet().iterator();
		 while(true)
		 {
			 
			 
	         String sid=(String)it2.next();	
			 stmt.execute("SELECT id,user_id,text,in_reply_to_user_id FROM status ;");
			
			 System.out.println("SELECT id,user_id,text,in_reply_to_user_id FROM status where id='"+sid+"';");
			 System.out.println("start query "); 
			rs = stmt.getResultSet();
			System.out.println("done query "); 
		 

		
		// Now do something with the ResultSet ....
		 // convert resultset to instances
			System.out.println("converting resultset to instances");
			
			if(rs!=null)
			{
		
			 // 3. fill with data
	
			    
			    
			    int num=0;
			    while(rs.next())
			    {
				    
//			    	System.out.println("num0="+num);
			    	/**
					 * id name location description  url
					 */
			    	
			    	Iterator it1=attributelist.iterator();
					while(it1.hasNext())
					{
						String attstr=(String)it1.next();
						String rows = null;
						
						rows=rs.getString(attstr.toLowerCase());
//						System.out.println(rows);
						ModelAttribute attribute=(ModelAttribute)hm.get(attstr);
						String attributename=attribute.getAttributeName();
						inhm.put(attributename.trim(), rows);
					}
				    Instance newInst = new Instance(data.numAttributes());
				    String labelfornewinst="missing";
				    
				    
					for(int i=0;i<data.numAttributes();i++)
					{
		//				System.out.println("i="+data.numAttributes());
						if(inhm.containsKey(data.attribute(i).name()))
						{
							if(inhm.get(data.attribute(i).name())==null)
								continue;
							
							if(data.attribute(i).name().equalsIgnoreCase("id"))
							{
								String key=(String)inhm.get(data.attribute(i).name());
								if(this.messagespamhm.containsKey(key))
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "spam");
									labelfornewinst="spam";
									/**
									 * according to the message id to obtain the userid/senderid
									 */
								
									continue;
//									num++;
								}
								else
								{
									newInst.setValue(data.attribute(data.numAttributes()-1), "ham");
									labelfornewinst="ham";
									num++;
								}
							}
							
							
						 //-numeric
						   if(data.attribute(i).isNumeric())
						   {
							   newInst.setValue(data.attribute(i),Double.valueOf((String) inhm.get(data.attribute(i).name())));
							   
						   }
						   //-nominal
						   if(data.attribute(i).isNominal())
						   {
							   MappingAttribute mapa=(MappingAttribute) mappingattrhm.get(data.attribute(i).name());
							   ModelAttribute modela=(ModelAttribute) hm.get(mapa.getAttributeName());
							   String tempstr=(String) inhm.get(data.attribute(i).name());
							   ArrayList modelal=modela.getNomialValues();
							   ArrayList mappingal=mapa.getNomialValues();
							   HashMap mappingnominal=new HashMap();
							   for(int n=0;n<modelal.size();n++)
							   {
//								   System.out.println("mapping:"+mappingal.get(n)+ modelal.get(n));
								   mappingnominal.put(mappingal.get(n), modelal.get(n));
							   }
							   
							   newInst.setValue(data.attribute(i),(String)mappingnominal.get(tempstr));
						   }
						   //-string
						   if(data.attribute(i).isString())
						   {
		//					   System.out.println("--"+data.attribute(i).name());
							   String tempstr=(String)inhm.get(data.attribute(i).name());
	//						   System.out.println(tempstr);
							   
							   newInst.setValue(data.attribute(i),tempstr);
								
		//					   vals[i] = data.attribute(i).addStringValue("This is a string!");
						   }
					   
						   //-date
						   if(data.attribute(i).isDate())
						   {
							   newInst.setValue(data.attribute(i),(String) inhm.get(data.attribute(i).name()) );
								
		//					   try {
		//						vals[i] = data.attribute(i).parseDate("2001-11-09");
		//					} catch (ParseException e) {
		//						// TODO Auto-generated catch block
		//						e.printStackTrace();
		//					}
						   }
						}
					}
					
					int id=totalnum+num;
					String instanceidstr="Twitter_"+id;
					
					newInst.setValue(data.attribute(data.numAttributes()-2), instanceidstr);
					
				    // add
				    data.add(newInst);
				    


			    	 //append new instance to the file
			    	
			    	writer =new FileWriter(arfffilename, true);
			    	
				    bw=new BufferedWriter(writer);
				    try {
				    	
				    	  // You are now at the end of the file,
				    	  // and can start writing new data out, e.g.
				    	 bw.newLine(); 
				    	 for(int i=0;i<data.numInstances();i++)
				    	 {
				    		 bw.write(data.instance(i).toString());
				    		 bw.newLine();
				    		 bw.flush();
				    	 }
				    	 bw.close();
				    	 writer.close();
				    	 bw=null;
				    	 writer=null;
				    	} catch (IOException ex ) {
				    	  ex.printStackTrace();
				    	}finally {                       // always close the file
				    		 if (bw != null) try {
				    		    bw.close();
				    		 } catch (IOException ioe2) {
				    		    // just ignore it
				    		 }
				    	 }
				    	// delete instances in data
				    data.delete();
				    
				    
				    
			
			    }
			    System.out.println("num="+num);
			    totalnum=totalnum+num;
			    System.out.println("totalnum="+totalnum);
			    if(totalnum==1500)
			    	break;
			   
			 // 4. output data
	//		    System.out.println(data);
			    
			   
			}
			
			
		  
		   
		    // first time saving 
		 
		    	
		  
		   
		    
		    
//		    Runtime.getRuntime().gc();
			offsetnum=offsetnum+limit;
		 }
		
		
		}
		catch (SQLException ex){
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		finally {
		// it is a good idea to release
		// resources in a finally{} block
		// in reverse-order of their creation
		// if they are no-longer needed
			if (rs != null) {
				try {
					rs.close();
				} catch (SQLException sqlEx) { } // ignore
					rs = null;
			}
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException sqlEx) { } // ignore
					stmt = null;
			}
		}
		
		} catch (SQLException ex) {
		// handle any errors
		System.out.println("SQLException: " + ex.getMessage());
		System.out.println("SQLState: " + ex.getSQLState());
		System.out.println("VendorError: " + ex.getErrorCode());
		}
		
		return data;
	}

	
	
/**
 * attributes mapping
 * @param attr
 * @return
 */
	public HashMap personMapping(ArrayList attr)
	{
		
		HashMap hm=new HashMap();
		Iterator it=attr.iterator();
		while(it.hasNext())
		{
			String tempstr=(String) it.next();
//			System.out.println("personMapping tempstr="+tempstr);
			/**
			 * id name location description  url
			 */
			MappingAttribute maa=(MappingAttribute)(this.personmappinghm.get(tempstr));
			String modelattributename=maa.getModelAttributeName();
//			System.out.println("modelattributename:"+modelattributename);
			ModelAttribute ma=(ModelAttribute)this.personmodelhm.get(modelattributename);

//			System.out.println("ma.getattributename:"+ma.getAttributeName());
			hm.put(tempstr,ma);
		}
		return hm;
		
	}
	
	
	/**
	 * attributes mapping
	 * @return
	 */
	public HashMap messageMapping(ArrayList attr)
	{

		HashMap hm=new HashMap();
		Iterator it=attr.iterator();
		while(it.hasNext())
		{
			String tempstr=(String) it.next();
//			System.out.println("personMapping tempstr="+tempstr);
			/**
			 * id, user_id, text, in_reply_to_user_id
			 */
			MappingAttribute maa=(MappingAttribute)(this.messagemappinghm.get(tempstr));
			String modelattributename=maa.getModelAttributeName();
//			System.out.println("modelattributename:"+modelattributename);
			ModelAttribute ma=(ModelAttribute)this.messagemodelhm.get(modelattributename);

//			System.out.println("ma.getattributename:"+ma.getAttributeName());
			hm.put(tempstr,ma);
		}
		return hm;
		


	}
	


	
	/**
	 * save instances to arff files
	 * @param in
	 * @param filedirectory
	 * @param filename
	 */
	public void save2arff(Instances in, String filedirectory, String filename)
	{
		 Instances dataSet =in;
		 ArffSaver saver = new ArffSaver();
		 saver.setInstances(dataSet);
		 try {
			saver.setFile(new File(filedirectory+System.getProperties().getProperty("file.separator")+filename));
		//	saver.setDestination(new File("./data/test.arff"));   // **not** necessary in 3.5.4 and later
			 saver.writeBatch();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		 
	}
	
	
	
	public int getIsappended() {
		return isappended;
	}


	public void setIsappended(int isappended) {
		this.isappended = isappended;
	}


	public static void  main(String [] arg0)
	{
		
		TwitterSpamHamFilter tshf=new TwitterSpamHamFilter();
		tshf.init();
	
//		Instances data=tshf.parseMessages(null);
		Instances data2=tshf.parsePersons(null);
//		tm.save2arff(data, System.getProperties().getProperty("user.dir"), "test.arff");
	}
}
